package com.zenitera.bigdata.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * 联合列表状态（ Union list state ）
 * 也是将状态表示为数据的列表。它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复
 * getUnionListState
 */
public class Flink01_State_Operator_UnionList {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        // 开启checkpoint
        env.enableCheckpointing(2000);
        env
                .socketTextStream("localhost", 6666)
                .map(new MyMapFunction())
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public static class MyMapFunction implements MapFunction<String, String>, CheckpointedFunction {

        List<String> words = new ArrayList<>();
        private ListState<String> wordsState;

        @Override
        public String map(String line) throws Exception {
            if (line.contains("exit")) {
                throw new RuntimeException("异常");
            }

            String[] data = line.split(" ");
            words.addAll(Arrays.asList(data));
            return words.toString();
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {

            wordsState.update(words);

        }

        @Override
        public void initializeState(FunctionInitializationContext ctx) throws Exception {
            wordsState = ctx.getOperatorStateStore().getUnionListState(new ListStateDescriptor<String>("wordsState", String.class));

            Iterable<String> it = wordsState.get();
            for (String word : it) {
                words.add(word);
            }

        }
    }
}

/*
D:\BaiduNetdiskDownload\3.安装包\netcat-win32-1.12>nc64.exe -lp 6666
aaa bbb ccc
ddd
exit

D:\BaiduNetdiskDownload\3.安装包\netcat-win32-1.12>nc64.exe -lp 6666
ggg
ppp aaaa
exit

D:\BaiduNetdiskDownload\3.安装包\netcat-win32-1.12>nc64.exe -lp 6666
zzz
-------------------------------------------
1> [aaa, bbb, ccc]
2> [ddd]
2> [ddd, aaa, bbb, ccc, ggg]
1> [ddd, aaa, bbb, ccc, ppp, aaaa]
1> [ddd, aaa, bbb, ccc, ggg, ddd, aaa, bbb, ccc, ppp, aaaa, zzz]
 */